package com.cctc.rds.scheduler.client.worker;

import com.cctc.rds.scheduler.client.worker.impl.DefaultTaskWorker;
import com.cctc.rds.scheduler.core.raft.TaskRpc;
import com.cctc.rds.scheduler.core.task.TaskLog;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;

/**
 * 任务（自动）执行调度器。
 * 描述信息：任务调度器客户端自动从 Leader 节点的执行容器中拉取任务并执行，执行完成后向 Leader 节点更新任务执行结果（任务状态：success、fail），并将
 * 更新后的任务实例日志（任务状态：success、fail）同步至所有 Follower 节点；
 * --> 如果任务实例日志状态为 success（执行成功），则结束执行。
 * --> 如果任务实例日志状态为 fail（执行失败），则设置重新调度时间为当前时间，并计算下一次延迟执行时间，更新任务状态为 pending。
 */
public class TaskWokerScheduler implements Runnable, TaskWorkerCallback, Closeable {
    private String app;
    private TaskRpc taskRpc;
    private TaskWorker taskWorker;
    private Thread taskWorkerScheduler;
    volatile private boolean status = true;

    public TaskWokerScheduler(String app, TaskRpc taskRpc) {
        this.app = app;
        this.taskRpc = taskRpc;
        this.taskWorker = new DefaultTaskWorker(app, Runtime.getRuntime().availableProcessors() * 2);
        this.taskWorkerScheduler = new Thread(this, "task-run-worker-scheduler");
        this.taskWorkerScheduler.setDaemon(true);
        this.taskWorkerScheduler.start();
    }

    @Override
    public void run() {
        final int maxRecords = 1000;
        long beginTime = System.currentTimeMillis() / 1000;
        while (status) {
            try {
                List<TaskLog> taskLogs = taskRpc.pull(app, maxRecords);
                if (!taskLogs.isEmpty()) {
                    taskLogs.parallelStream().forEach(taskLog -> taskWorker.run(taskLog, this));
                }
                long curTime = System.currentTimeMillis() / 1000;
                if (taskLogs.size() < maxRecords && curTime <= beginTime) {
                    long waitTime = curTime % 1000;
                    if (waitTime > 0) Thread.sleep(waitTime);
                }
            } catch (Exception e) {
            }
        }
    }

    @Override
    public void onFinish(long taskId, boolean success, Throwable e) {
        taskRpc.commit(app, taskId, success);
        if (e != null) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() throws IOException {
        status = false;
        taskWorker.close();
    }

}
